{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.context import SparkContext\n",
    "from pyspark.sql.session import SparkSession"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [],
   "source": [
    "sc = SparkContext(\"local[*]\",\"Forest Cover Type\")\n",
    "spark = SparkSession(sc)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "'0,tcp,http,SF,181,5450,0,0,0,0,0,1,0,0,0,0,0,0,0,0,0,0,8,8,0.00,0.00,0.00,0.00,1.00,0.00,0.00,9,9,1.00,0.00,0.11,0.00,0.00,0.00,0.00,0.00,normal.'"
      ]
     },
     "execution_count": 3,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "kddCup = sc.textFile(\"data/KddCup/kddcup.data_10_percent_corrected\")\n",
    "first = kddCup.first()\n",
    "first"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "42"
      ]
     },
     "execution_count": 4,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "len(first.split(\",\"))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "['duration',\n",
       " 'protocol_type',\n",
       " 'service',\n",
       " 'flag',\n",
       " 'src_bytes',\n",
       " 'dst_bytes',\n",
       " 'land',\n",
       " 'wrong_fragment',\n",
       " 'urgent',\n",
       " 'hot',\n",
       " 'num_failed_logins',\n",
       " 'logged_in',\n",
       " 'num_compromised',\n",
       " 'root_shell',\n",
       " 'su_attempted',\n",
       " 'num_root',\n",
       " 'num_file_creations',\n",
       " 'num_shells',\n",
       " 'num_access_files',\n",
       " 'num_outbound_cmds',\n",
       " 'is_host_login',\n",
       " 'is_guest_login',\n",
       " 'count',\n",
       " 'srv_count',\n",
       " 'serror_rate',\n",
       " 'srv_serror_rate',\n",
       " 'rerror_rate',\n",
       " 'srv_rerror_rate',\n",
       " 'same_srv_rate',\n",
       " 'diff_srv_rate',\n",
       " 'srv_diff_host_rate',\n",
       " 'dst_host_count',\n",
       " 'dst_host_srv_count',\n",
       " 'dst_host_same_srv_rate',\n",
       " 'dst_host_diff_srv_rate',\n",
       " 'dst_host_same_src_port_rate',\n",
       " 'dst_host_srv_diff_host_rate',\n",
       " 'dst_host_serror_rate',\n",
       " 'dst_host_srv_serror_rate',\n",
       " 'dst_host_rerror_rate',\n",
       " 'dst_host_srv_rerror_rate',\n",
       " 'label']"
      ]
     },
     "execution_count": 5,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "header = open(\"data/KddCup/header.txt\").read().split(\"\\n\")\n",
    "header = [x.split(\":\")[0] for x in header][:-1]\n",
    "header"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "42"
      ]
     },
     "execution_count": 6,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "len(header)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [],
   "source": [
    "kddCup = kddCup.map(lambda row : row.split(\",\"))\n",
    "kddDataFrame = spark.createDataFrame(kddCup,header)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------+-------------+-------+----+-------+\n",
      "|duration|protocol_type|service|flag|  label|\n",
      "+--------+-------------+-------+----+-------+\n",
      "|       0|          tcp|   http|  SF|normal.|\n",
      "|       0|          tcp|   http|  SF|normal.|\n",
      "|       0|          tcp|   http|  SF|normal.|\n",
      "|       0|          tcp|   http|  SF|normal.|\n",
      "|       0|          tcp|   http|  SF|normal.|\n",
      "|       0|          tcp|   http|  SF|normal.|\n",
      "|       0|          tcp|   http|  SF|normal.|\n",
      "|       0|          tcp|   http|  SF|normal.|\n",
      "|       0|          tcp|   http|  SF|normal.|\n",
      "|       0|          tcp|   http|  SF|normal.|\n",
      "|       0|          tcp|   http|  SF|normal.|\n",
      "|       0|          tcp|   http|  SF|normal.|\n",
      "|       0|          tcp|   http|  SF|normal.|\n",
      "|       0|          tcp|   http|  SF|normal.|\n",
      "|       0|          tcp|   http|  SF|normal.|\n",
      "|       0|          tcp|   http|  SF|normal.|\n",
      "|       0|          tcp|   http|  SF|normal.|\n",
      "|       0|          tcp|   http|  SF|normal.|\n",
      "|       0|          tcp|   http|  SF|normal.|\n",
      "|       0|          tcp|   http|  SF|normal.|\n",
      "+--------+-------------+-------+----+-------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "kddDataFrame.select(header[0],header[1],header[2],\n",
    "                    header[3],header[-1]).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- duration: string (nullable = true)\n",
      " |-- protocol_type: string (nullable = true)\n",
      " |-- service: string (nullable = true)\n",
      " |-- flag: string (nullable = true)\n",
      " |-- src_bytes: string (nullable = true)\n",
      " |-- dst_bytes: string (nullable = true)\n",
      " |-- land: string (nullable = true)\n",
      " |-- wrong_fragment: string (nullable = true)\n",
      " |-- urgent: string (nullable = true)\n",
      " |-- hot: string (nullable = true)\n",
      " |-- num_failed_logins: string (nullable = true)\n",
      " |-- logged_in: string (nullable = true)\n",
      " |-- num_compromised: string (nullable = true)\n",
      " |-- root_shell: string (nullable = true)\n",
      " |-- su_attempted: string (nullable = true)\n",
      " |-- num_root: string (nullable = true)\n",
      " |-- num_file_creations: string (nullable = true)\n",
      " |-- num_shells: string (nullable = true)\n",
      " |-- num_access_files: string (nullable = true)\n",
      " |-- num_outbound_cmds: string (nullable = true)\n",
      " |-- is_host_login: string (nullable = true)\n",
      " |-- is_guest_login: string (nullable = true)\n",
      " |-- count: string (nullable = true)\n",
      " |-- srv_count: string (nullable = true)\n",
      " |-- serror_rate: string (nullable = true)\n",
      " |-- srv_serror_rate: string (nullable = true)\n",
      " |-- rerror_rate: string (nullable = true)\n",
      " |-- srv_rerror_rate: string (nullable = true)\n",
      " |-- same_srv_rate: string (nullable = true)\n",
      " |-- diff_srv_rate: string (nullable = true)\n",
      " |-- srv_diff_host_rate: string (nullable = true)\n",
      " |-- dst_host_count: string (nullable = true)\n",
      " |-- dst_host_srv_count: string (nullable = true)\n",
      " |-- dst_host_same_srv_rate: string (nullable = true)\n",
      " |-- dst_host_diff_srv_rate: string (nullable = true)\n",
      " |-- dst_host_same_src_port_rate: string (nullable = true)\n",
      " |-- dst_host_srv_diff_host_rate: string (nullable = true)\n",
      " |-- dst_host_serror_rate: string (nullable = true)\n",
      " |-- dst_host_srv_serror_rate: string (nullable = true)\n",
      " |-- dst_host_rerror_rate: string (nullable = true)\n",
      " |-- dst_host_srv_rerror_rate: string (nullable = true)\n",
      " |-- label: string (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "kddDataFrame.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----------------+------+\n",
      "|           label| count|\n",
      "+----------------+------+\n",
      "|    warezmaster.|    20|\n",
      "|          smurf.|280790|\n",
      "|            pod.|   264|\n",
      "|           imap.|    12|\n",
      "|           nmap.|   231|\n",
      "|   guess_passwd.|    53|\n",
      "|        ipsweep.|  1247|\n",
      "|      portsweep.|  1040|\n",
      "|          satan.|  1589|\n",
      "|           land.|    21|\n",
      "|     loadmodule.|     9|\n",
      "|      ftp_write.|     8|\n",
      "|buffer_overflow.|    30|\n",
      "|        rootkit.|    10|\n",
      "|    warezclient.|  1020|\n",
      "|       teardrop.|   979|\n",
      "|           perl.|     3|\n",
      "|            phf.|     4|\n",
      "|       multihop.|     7|\n",
      "|        neptune.|107201|\n",
      "|           back.|  2203|\n",
      "|            spy.|     2|\n",
      "|         normal.| 97278|\n",
      "+----------------+------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "kddDataFrame.groupby(\"label\").count().show(23)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[Row(duration='0', protocol_type='tcp', service='http', flag='SF', src_bytes='181', dst_bytes='5450', land='0', wrong_fragment='0', urgent='0', hot='0', num_failed_logins='0', logged_in='1', num_compromised='0', root_shell='0', su_attempted='0', num_root='0', num_file_creations='0', num_shells='0', num_access_files='0', num_outbound_cmds='0', is_host_login='0', is_guest_login='0', count='8', srv_count='8', serror_rate='0.00', srv_serror_rate='0.00', rerror_rate='0.00', srv_rerror_rate='0.00', same_srv_rate='1.00', diff_srv_rate='0.00', srv_diff_host_rate='0.00', dst_host_count='9', dst_host_srv_count='9', dst_host_same_srv_rate='1.00', dst_host_diff_srv_rate='0.00', dst_host_same_src_port_rate='0.11', dst_host_srv_diff_host_rate='0.00', dst_host_serror_rate='0.00', dst_host_srv_serror_rate='0.00', dst_host_rerror_rate='0.00', dst_host_srv_rerror_rate='0.00', label='normal.')]"
      ]
     },
     "execution_count": 11,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "kddDataFrame.take(1)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {},
   "outputs": [],
   "source": [
    "kddDataFrame = kddDataFrame.drop(\"protocol_type\",\"service\",\"flag\",\"label\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- duration: string (nullable = true)\n",
      " |-- src_bytes: string (nullable = true)\n",
      " |-- dst_bytes: string (nullable = true)\n",
      " |-- land: string (nullable = true)\n",
      " |-- wrong_fragment: string (nullable = true)\n",
      " |-- urgent: string (nullable = true)\n",
      " |-- hot: string (nullable = true)\n",
      " |-- num_failed_logins: string (nullable = true)\n",
      " |-- logged_in: string (nullable = true)\n",
      " |-- num_compromised: string (nullable = true)\n",
      " |-- root_shell: string (nullable = true)\n",
      " |-- su_attempted: string (nullable = true)\n",
      " |-- num_root: string (nullable = true)\n",
      " |-- num_file_creations: string (nullable = true)\n",
      " |-- num_shells: string (nullable = true)\n",
      " |-- num_access_files: string (nullable = true)\n",
      " |-- num_outbound_cmds: string (nullable = true)\n",
      " |-- is_host_login: string (nullable = true)\n",
      " |-- is_guest_login: string (nullable = true)\n",
      " |-- count: string (nullable = true)\n",
      " |-- srv_count: string (nullable = true)\n",
      " |-- serror_rate: string (nullable = true)\n",
      " |-- srv_serror_rate: string (nullable = true)\n",
      " |-- rerror_rate: string (nullable = true)\n",
      " |-- srv_rerror_rate: string (nullable = true)\n",
      " |-- same_srv_rate: string (nullable = true)\n",
      " |-- diff_srv_rate: string (nullable = true)\n",
      " |-- srv_diff_host_rate: string (nullable = true)\n",
      " |-- dst_host_count: string (nullable = true)\n",
      " |-- dst_host_srv_count: string (nullable = true)\n",
      " |-- dst_host_same_srv_rate: string (nullable = true)\n",
      " |-- dst_host_diff_srv_rate: string (nullable = true)\n",
      " |-- dst_host_same_src_port_rate: string (nullable = true)\n",
      " |-- dst_host_srv_diff_host_rate: string (nullable = true)\n",
      " |-- dst_host_serror_rate: string (nullable = true)\n",
      " |-- dst_host_srv_serror_rate: string (nullable = true)\n",
      " |-- dst_host_rerror_rate: string (nullable = true)\n",
      " |-- dst_host_srv_rerror_rate: string (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "kddDataFrame.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.sql.types import DoubleType\n",
    "for x in kddDataFrame.columns:\n",
    "    kddDataFrame = kddDataFrame.withColumn(x, \n",
    "                                           kddDataFrame[x].cast(DoubleType()))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[Row(duration=0.0, src_bytes=181.0, dst_bytes=5450.0, land=0.0, wrong_fragment=0.0, urgent=0.0, hot=0.0, num_failed_logins=0.0, logged_in=1.0, num_compromised=0.0, root_shell=0.0, su_attempted=0.0, num_root=0.0, num_file_creations=0.0, num_shells=0.0, num_access_files=0.0, num_outbound_cmds=0.0, is_host_login=0.0, is_guest_login=0.0, count=8.0, srv_count=8.0, serror_rate=0.0, srv_serror_rate=0.0, rerror_rate=0.0, srv_rerror_rate=0.0, same_srv_rate=1.0, diff_srv_rate=0.0, srv_diff_host_rate=0.0, dst_host_count=9.0, dst_host_srv_count=9.0, dst_host_same_srv_rate=1.0, dst_host_diff_srv_rate=0.0, dst_host_same_src_port_rate=0.11, dst_host_srv_diff_host_rate=0.0, dst_host_serror_rate=0.0, dst_host_srv_serror_rate=0.0, dst_host_rerror_rate=0.0, dst_host_srv_rerror_rate=0.0)]"
      ]
     },
     "execution_count": 15,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "kddDataFrame.take(1)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.ml.clustering import KMeans\n",
    "assembler = VectorAssembler(\n",
    "        inputCols=kddDataFrame.columns,\n",
    "        outputCol=\"features\")\n",
    "output = assembler.setHandleInvalid(\"skip\").transform(kddDataFrame)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------------+\n",
      "|            features|\n",
      "+--------------------+\n",
      "|(38,[1,2,8,19,20,...|\n",
      "|(38,[1,2,8,19,20,...|\n",
      "|(38,[1,2,8,19,20,...|\n",
      "|(38,[1,2,8,19,20,...|\n",
      "|(38,[1,2,8,19,20,...|\n",
      "|(38,[1,2,8,19,20,...|\n",
      "|(38,[1,2,8,19,20,...|\n",
      "|(38,[1,2,8,19,20,...|\n",
      "|(38,[1,2,8,19,20,...|\n",
      "|(38,[1,2,6,8,19,2...|\n",
      "|(38,[1,2,8,19,20,...|\n",
      "|(38,[1,2,8,19,20,...|\n",
      "|(38,[1,2,8,19,20,...|\n",
      "|(38,[1,2,8,19,20,...|\n",
      "|(38,[1,2,8,19,20,...|\n",
      "|(38,[1,2,8,19,20,...|\n",
      "|(38,[1,2,8,19,20,...|\n",
      "|(38,[1,2,8,19,20,...|\n",
      "|(38,[1,2,8,19,20,...|\n",
      "|(38,[1,2,8,19,20,...|\n",
      "+--------------------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "output.select(\"features\").show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 24,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.ml.feature import VectorAssembler\n",
    "kmeans = KMeans()\n",
    "kmeans_model = kmeans.fit(output.select(\"features\"))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 25,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "number of centers:2\n",
      "Cluster centers: \n",
      "[array([4.79793956e+01, 1.62207883e+03, 8.68534183e+02, 4.45326100e-05,\n",
      "       6.43293794e-03, 1.41694668e-05, 3.45168212e-02, 1.51815716e-04,\n",
      "       1.48247035e-01, 1.02121372e-02, 1.11331525e-04, 3.64357718e-05,\n",
      "       1.13517671e-02, 1.08295211e-03, 1.09307315e-04, 1.00805635e-03,\n",
      "       0.00000000e+00, 0.00000000e+00, 1.38658354e-03, 3.32286248e+02,\n",
      "       2.92907143e+02, 1.76685418e-01, 1.76607809e-01, 5.74330999e-02,\n",
      "       5.77183920e-02, 7.91548844e-01, 2.09816404e-02, 2.89968625e-02,\n",
      "       2.32470732e+02, 1.88666046e+02, 7.53781203e-01, 3.09056111e-02,\n",
      "       6.01935529e-01, 6.68351484e-03, 1.76753957e-01, 1.76441622e-01,\n",
      "       5.81176268e-02, 5.74111170e-02]), array([2.0000000e+00, 6.9337564e+08, 0.0000000e+00, 0.0000000e+00,\n",
      "       0.0000000e+00, 0.0000000e+00, 1.0000000e+00, 0.0000000e+00,\n",
      "       0.0000000e+00, 0.0000000e+00, 0.0000000e+00, 0.0000000e+00,\n",
      "       0.0000000e+00, 0.0000000e+00, 0.0000000e+00, 0.0000000e+00,\n",
      "       0.0000000e+00, 0.0000000e+00, 0.0000000e+00, 5.7000000e+01,\n",
      "       3.0000000e+00, 7.9000000e-01, 6.7000000e-01, 2.1000000e-01,\n",
      "       3.3000000e-01, 5.0000000e-02, 3.9000000e-01, 0.0000000e+00,\n",
      "       2.5500000e+02, 3.0000000e+00, 1.0000000e-02, 9.0000000e-02,\n",
      "       2.2000000e-01, 0.0000000e+00, 1.8000000e-01, 6.7000000e-01,\n",
      "       5.0000000e-02, 3.3000000e-01])]\n"
     ]
    }
   ],
   "source": [
    "# Shows the result\n",
    "centers = kmeans_model.clusterCenters()\n",
    "print(\"number of centers:{}\".format(len(centers)))\n",
    "print(\"Cluster centers: \")\n",
    "print(centers)\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.7.6"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}
